package raft


//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"math/rand"
	"sync/atomic"
	"time"
	"os"
	"log"
	"fmt"
	"6.5840/labgob"
	"6.5840/labrpc"
)


// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 3D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.

const (
	FOLLOWER = 0
	CANDIDATE = 1
	LEADER = 2
)

const (
	ElectionTimeoutMin = 150
	ElectionTimeoutMax = 350
	KeepAlivePeriod = 50
)

type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 3D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type LogEntry struct {
	Index	int
	Term	int
	Command	interface{}
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        SyncMutex          // Lock to protect shared access to 
				      // this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted
				      // state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	//Raft paper figure2: Persistent state on all servers
	CurrentTerm  int
	VotedFor     int
	Log	     []LogEntry

	//Raft paper figure2: Volatile state on all servers
	CommitIndex  int
	LastApplied  int

	//Raft paper figure2: Volatile state on leaders
	NextIndex   []int
	MatchIndex  []int


	//这些不是在Raft论文里面定义的
	ApplyCh   chan ApplyMsg
	State     int

	//收到了多少次keepalive, 所有的AE消息都算做keepalive
	NumKeepAlive	int

	//发送了多少次AppendEntries消息
	NumAppendEntriesSent	int

	//谁给我投了票
	VotedForMe	[]int

	//用来记录发送AppendEntries消息的序列号
	aeTxSeq		[]int
	aeRxSeq		[]int

	//是否正在睡眠
	sleeping bool

	//定时器
	Timer           *time.Timer

	//for 3D
	LastsnapIndex int
	LastsnapTerm int
	Snap []byte


}

func Min(x, y int) int {
	if x < y {
		return x
	}
	return y
}

func Max(x, y int) int {
	if x > y {
		return x
	}
	return y
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	// Your code here (3A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	isleader := false
	if rf.State == LEADER {
		isleader = true
	}
	rf.Dbg("isleader=%v\n", isleader)
	return rf.CurrentTerm, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (3C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.CurrentTerm)
	e.Encode(rf.VotedFor)
	e.Encode(rf.Log)

	e.Encode(rf.LastsnapIndex)
	e.Encode(rf.LastsnapTerm)

	raftstate := w.Bytes()
	rf.persister.Save(raftstate, nil)
	rf.Dbg("persist\n")
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (3C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var term int
	var votedFor int
	var LastsnapIndex int
	var LastsnapTerm int
	log := make([]LogEntry, 0)
	ok := true
	if d.Decode(&term) != nil ||
		d.Decode(&votedFor) != nil ||
		d.Decode(&log) != nil ||
		d.Decode(&LastsnapIndex)!=nil||
		d.Decode(&LastsnapTerm)!= nil{
		ok = false

	} else {
		rf.CurrentTerm = term
		rf.VotedFor = votedFor
		rf.Log = log
		rf.LastsnapIndex=LastsnapIndex
		rf.LastsnapTerm=LastsnapTerm
		rf.CommitIndex=LastsnapIndex
		rf.LastApplied=LastsnapIndex

	}
	rf.Dbg("ok=%v, log=%s\n", ok, SprintLog(log))
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// r := bytes.NewBuffer(snapshot)
	//d := labgob.NewDecoder(r)
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if index>rf.CommitIndex {
		return
	}
	if index<=rf.LastsnapIndex {
		return
	}
	rf.Snap=snapshot
	//rf.LastsnapIndex=index
	rf.LastsnapTerm=rf.Log[rf.findid(index)].Term
	rf.Log=rf.Log[rf.findid(index):]
	rf.LastsnapIndex=index
	if rf.LastApplied<index{
		rf.LastApplied=index
	}
	rf.persist()
	return

// Your code here (3D).
}
func (rf *Raft) findid(index int) int{
	return index-rf.LastsnapIndex

}
func (rf *Raft) reachid(index int) int{
	return index+ rf.LastsnapIndex
}


func (rf *Raft) ReadSnapshot(data []byte){
	if len(data) == 0 {
		return
	}
	rf.Snap = data
	return
}
type InstallsnapshotArgs struct{
	Term int
	Leaderid int 
	LastsnapIndex int
	LastsnapTerm int
	Data []byte
	LastsnapCmd int


}

type InstallsnapshotReply struct{
	Term int
}
func (rf *Raft) handleInstallsnap(serverTo int){

	args:=InstallsnapshotArgs{
		Term : rf.CurrentTerm,
		Leaderid : rf.me,
		LastsnapIndex:rf.LastsnapIndex,
		LastsnapTerm : rf.LastsnapTerm,
	}
	reply:= InstallsnapshotReply{}

	rf.mu.Lock()
	defer rf.mu.Unlock()

	if(rf.State!=LEADER){
		//rf.mu.Unlock()
		return
	}

	ok:= rf.sendInstallsnap(serverTo,&args,&reply)
	if(!ok){
		//rf.mu.Unlock()
		return
	}
	if(rf.CurrentTerm<reply.Term){
		rf.CurrentTerm=reply.Term
		rf.State=FOLLOWER
		rf.VotedFor=-1
		//rf.mu.Unlock()
		rf.persist()
	}

	return

}//发送快照，leader方处理
func (rf *Raft) Installsnap(args *InstallsnapshotArgs,reply *InstallsnapshotReply){
	rf.mu.Lock()
	defer rf.mu.Unlock()

	reply.Term = rf.CurrentTerm

	if(rf.CurrentTerm>args.Term){
		return
	}

	if(rf.CurrentTerm<args.Term){
		rf.CurrentTerm = args.Term
		rf.VotedFor = -1
		rf.persist()
	}

	rf.State = FOLLOWER

	idx := 0
	hasEntry := false
	for ;idx<len(rf.Log);idx++{
		if rf.reachid(idx)==args.LastsnapIndex && rf.LastsnapTerm == args.LastsnapTerm{
			hasEntry = true
			break
		}
	}
	if(hasEntry){
		rf.Log=rf.Log[idx:]
	}else{
		rf.Log=make([]LogEntry,0)
		rf.Log=append(rf.Log,LogEntry{Term: args.LastsnapTerm,Index:args.LastsnapIndex})
	}

	rf.Snap = args.Data
	rf.LastsnapIndex = args.LastsnapIndex
	rf.LastsnapTerm = rf.LastsnapTerm

	rf.CommitIndex = Max(rf.CommitIndex,rf.LastsnapIndex)
	rf.LastApplied = Max(rf.LastApplied,rf.LastsnapIndex)
	reply.Term = rf.CurrentTerm
	return
}


// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (3A, 3B).
	Term int
	CandidateId int
	LastLogIndex int
	LastLogTerm int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (3A).
	Term int
	VoteGranted bool
}

type AppendEntriesArgs struct {
	Seq int
	Term int
	LeaderId int
	PrevLogIndex int
	PrevLogTerm int
	Entries	[]LogEntry
	LeaderCommit int
}

type AppendEntriesReply struct {
	Seq int
	Term int
	LastIndex int
	Success bool
}

func (ae *AppendEntriesArgs)Sprint() string {
	s := fmt.Sprintf("{sq=%d, tm=%d, LdId=%d, PrLgIdx=%d, PrLgTm=%d, CmtIdx=%d, ",
		ae.Seq, ae.Term, ae.LeaderId, ae.PrevLogIndex, ae.PrevLogTerm,
		ae.LeaderCommit)
	n := len(ae.Entries)
	if n == 0 {
		s += "[]}"
	} else {
		s += SprintLog(ae.Entries)
		s += "}"
	}
	return s
}

//判断是否对方（RV消息的发送者）的日志比自己的更新
func (rf *Raft) isPeersLogUptodate(args *RequestVoteArgs) bool {

	//先用自己的最后一条日志的Term跟对方的对比
	lastEntry := rf.Log[len(rf.Log)-1]
	if lastEntry.Term > args.LastLogTerm {
		return false
	}

	if lastEntry.Term < args.LastLogTerm {
		return true
	}

	//如果自己最后一条日志的Term与对方相等，则比较谁的Log更多
	if rf.findid(len(rf.Log) - 1)> args.LastLogIndex {
		return false
	}

	return true
}


// 当接收到其他节点发送过来的RequestVote消息的时候，会调用本函数
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (3A, 3B).

	rf.Dbg("args = %#v\n", *args)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	//如果对方Term比自己的小，那么就拒绝给对方投票
	if args.Term < rf.CurrentTerm {
		reply.Term = rf.CurrentTerm
		reply.VoteGranted = false
		rf.Dbg("rejectVote, reply = %#v\n\t\targs=%#v\n",
			*reply, *args)
		return
	}

	//如果对方Term比自己的更大
	if args.Term > rf.CurrentTerm {
		//先把自己的Term更新为对方的Term
		rf.VotedFor = -1

		if rf.State != FOLLOWER {
			rf.becomeFollower(args.Term)
		} else {
			rf.changeTerm(args.Term)
		}

		//如果自己的Log比对方更uptodate，就决绝给对方投票；
		reply.Term = rf.CurrentTerm
		if rf.isPeersLogUptodate(args) {
			rf.VotedFor = args.CandidateId
			reply.VoteGranted = true
			rf.Dbg("acceptVote, reply = %#v\n\t\targs=%#v\n",
				*reply, *args)
			rf.persist()
		} else {
			reply.VoteGranted = false
			rf.Dbg("rejectVote, reply = %#v\n\t\targs=%#v\n",
				*reply, *args)
		}
		return
	}

	//下面是args.Term == rf.CurrentTerm的情况
	if rf.State == FOLLOWER {
		reply.Term = rf.CurrentTerm
		if (rf.VotedFor >= 0) && (rf.VotedFor != args.CandidateId) {
			//如果已经给其他节点投过票了，拒绝再给这个节点投票
			reply.VoteGranted = false
			rf.Dbg("rejectVote, reply = %#v\n\t\targs=%#v\n",
				*reply, *args)
		} else {
			if rf.isPeersLogUptodate(args) {
				rf.VotedFor = args.CandidateId
				reply.VoteGranted = true
				rf.Dbg("acceptVote, reply = %#v\n\t\targs=%#v\n",
					*reply, *args)
				rf.persist()
			} else {
				reply.VoteGranted = false
				rf.Dbg("rejectVote, reply = %#v\n\t\targs=%#v\n",
					*reply, *args)
			}
		}
	} else {
		//如果是Candidate或者Leader，对方Term跟自己一样，拒绝投票
		reply.Term = rf.CurrentTerm
		reply.VoteGranted = false
		rf.Dbg("rejectVote, reply = %#v\n\t\targs=%#v\n", 
			*reply, *args)
	}
}

//从index之后，全部剪掉。包括index
func (rf *Raft)cutOffLogTail(index int) {
	rf.Log = rf.Log[:index]
	rf.persist()
}

func (rf *Raft)appendEntryToSelf(entry *LogEntry) {
	rf.Log = append(rf.Log, *entry)
	rf.persist()
}

func (rf *Raft)appendEntriesToSelf(entries []LogEntry) {
	rf.Log = append(rf.Log, entries...)
	rf.persist()
}

// example RequestVote RPC handler.
func (rf *Raft) AppendEntries(args *AppendEntriesArgs,
		reply *AppendEntriesReply) {

	//申请锁，避免竞争访问
	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.Dbg("args=%s\n", args.Sprint())
	reply.Seq = args.Seq

	//对方的Term比我的更小，说明对方不应该是Leader
	if args.Term < rf.CurrentTerm {
		reply.Term = rf.CurrentTerm
		reply.Success = false
		reply.LastIndex = len(rf.Log) - 1
		rf.Dbg("reply=%#v\n\t\targs=%s\n", *reply, args.Sprint())
		return
	}

	if rf.State == LEADER {
		if args.Term > rf.CurrentTerm {
			//说明其他节点已经被选为新Term的Leader了
			//自己退位，变成Follower
			rf.VotedFor = -1
			rf.becomeFollower(args.Term)
		} else {
			//如果对方term跟自己的term相同，并且自己和对方都是leader
			//在同一个term里面存在2个leader，一定是有Bug
			rf.Dbg("BUG: Two leaders in the same term\n")
			os.Exit(3)
		}
	} else if rf.State == CANDIDATE {
		//说明在同一个Term或者更高Term其他节点，已经竞选成功，成为
		//leader，自己退位成为follower。
		if args.Term > rf.CurrentTerm {
			rf.VotedFor = -1
		}
		rf.becomeFollower(args.Term)
	} else if rf.State == FOLLOWER {
		if args.Term > rf.CurrentTerm {
			// 如果对方term比自己的大
			rf.VotedFor = -1
			rf.changeTerm(args.Term)
		} else {
			//如果对方term跟自己的相等
			rf.NumKeepAlive += 1
		}
	}

	//如果leader的前一个log entry的term跟自己相同位置的entry的term不匹配
	//则要求leader回退重发
	if (args.PrevLogIndex > 0) && (args.PrevLogIndex < len(rf.Log)) &&
	   rf.Log[args.PrevLogIndex].Term != args.PrevLogTerm {
		//如果每次回退1个，回退太慢了，无法通过测试
		 //所以改为每次回退一半
		backTo := Min(len(rf.Log)/2, args.PrevLogIndex)

		//已经提交了的entry，就不能被删除了
		if backTo < rf.CommitIndex {
			backTo = rf.CommitIndex
		}
		rf.cutOffLogTail(backTo+1)
		reply.LastIndex = backTo
		reply.Term = rf.CurrentTerm
		reply.Success = false
		rf.Dbg("reply=%#v\n\t\targs=%s\n", *reply, args.Sprint())
		return
	}

	//如果自己的日志少于PrevLogIndex，那么就返回false，要求下次AppendEntries
	//带更多旧的日志
	if args.PrevLogIndex > len(rf.Log) - 1 {
		reply.Term = rf.CurrentTerm
		reply.LastIndex = len(rf.Log) - 1
		reply.Success = false
		rf.Dbg("reply=%#v\n\t\targs=%s\n", *reply, args.Sprint())
		return
	}

	//如果自己的LogEntries多了，则截短
	if args.PrevLogIndex < len(rf.Log) - 1 {
		rf.cutOffLogTail(args.PrevLogIndex+1)
	}

	//1. 如果自己日志少，已经提前返回了；2. 如果自己日志多，已经截短了；
	//以上两种情况处理之后，自己的日志，应该i刚好跟args里面日志匹配上
	rf.assert(len(rf.Log)-1 == args.PrevLogIndex)

	//如果AE消息中携带了日志entries，那么把这些entries添加到自己的日志里面
	if len(args.Entries) > 0 {
		rf.appendEntriesToSelf(args.Entries)
	}

	rf.CommitIndex = args.LeaderCommit
	// 要把已经commit了的log entries apply到状态机，并更新LastApplied
	for i:=rf.LastApplied+1; i<=args.LeaderCommit; i++ {
		rf.Apply(i, rf.Log[i].Command)
	}

	reply.LastIndex = len(rf.Log) - 1
	reply.Term = rf.CurrentTerm
	reply.Success = true
	rf.Dbg("reply=%#v\n\t\targs=%s\n", *reply, args.Sprint())
}


// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.

func (rf *Raft) checkTimeout(t0 time.Time, limit int64) {
	t1 := time.Now()
	td := t1.Sub(t0)
	us := td.Microseconds()
	if us > limit {
		rf.Dbg("takes too long, %d us\n", us)
	}
}

func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs,
		reply *RequestVoteReply) bool {
	rf.Dbg("SendRPC.RV, to=%d, args=%#v\n", server, *args)
	t0 := time.Now()
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	if ok {
		rf.checkTimeout(t0, 4000)
		rf.Dbg("RcvRPL.RV, from=%d, ok=%v, reply=%#v\n\t\targs=%#v\n",
			server, ok, *reply, *args)
	}
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs,
		reply *AppendEntriesReply) bool {
	rf.Dbg("SendRPC.AE, to=%d, args=%s\n", server, args.Sprint())
	t0 := time.Now()
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	if ok {
		rf.checkTimeout(t0, 4000)
		rf.Dbg("RcvRPL.AE, from=%d, ok=%v, reply=%#v\n\t\targs=%s\n",
			server, ok, *reply, args.Sprint())
	}
	return ok
}
func (rf *Raft) sendInstallsnap(serverTo int,args *InstallsnapshotArgs,
		reply *InstallsnapshotReply) bool {
	t0 := time.Now()
	ok := rf.peers[serverTo].Call("Raft.Installsnap",args,reply)
	if ok{
		rf.checkTimeout(t0,4000)
	}
	return ok
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {

	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.State != LEADER || rf.killed() {
		return -1, -1, false
	}

	s := SprintCmd(command)
	rf.Dbg("cmd=%v\n", s)
	index := len(rf.Log)
	entry := LogEntry {Index: index, Term: rf.CurrentTerm, Command: command}

	rf.appendEntryToSelf(&entry)

	rf.NextIndex[rf.me] = index + 1
	rf.MatchIndex[rf.me] = index

	//唤醒goroutine leaderTick()处理这个LogEntry
	rf.wakeupTicker()

	// Your code here (3B).
	isLeader := true
	return index, rf.CurrentTerm, isLeader
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	rf.Dbg("I'm killed\n")
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) pleaseVoteForMe(to int) {
	//构造RequestVote消息，发给对方

	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.State != CANDIDATE {
		rf.Dbg("I am not a candidate\n")
		//rf.mu.Unlock()
		return
	}
	args := &RequestVoteArgs{}
	reply := &RequestVoteReply{}
	args.Term = rf.CurrentTerm
	args.CandidateId = rf.me
	args.LastLogIndex = len(rf.Log) - 1
	args.LastLogTerm = rf.Log[args.LastLogIndex].Term
	//rf.mu.Unlock()

	ok := rf.sendRequestVote(to, args, reply)

	//本函数跟startNewElection是在不同的goroutine中运行，因此需要重新申请锁
	//rf.mu.Lock()
	//defer rf.mu.Unlock()

	if !ok {
		rf.Dbg("Net failed while sending RV=%#v\n", args)
		return
	}

	//如果已经当选为Leader了，或者其他节点当选为Leader了，就不需要处理
	//RequestVote Reply消息了
	if rf.State != CANDIDATE {
		rf.Dbg("I'm not in candidate state\n")
		return
	}

	//如果对方的term比自己的大，那么自己应该变成Follower
	if reply.Term > rf.CurrentTerm {
		rf.Dbg("My term is lagged.\n")
		rf.VotedFor = -1
		rf.becomeFollower(reply.Term)
		return
	}

	//如果对方投票给自己了
	if reply.VoteGranted == true {
		//首先检查Term，因为网络消息传递可能发生延迟，这个投票消息可能
		//是很久以前其他term的。如果Term相同，则投票是有效的。如果term
		//不同，则说明对方term更小（前面已经检查过对方更大的情况了），
		//那么就忽略这个消息。
		if reply.Term == rf.CurrentTerm {
			rf.Dbg("Accepted to vote for me\n");
			rf.VotedForMe = append(rf.VotedForMe, to)
		} else {
			rf.Dbg("Ignore the vote for me reply sent by outdated server\n");
			return
		}
	} else {
		rf.Dbg("Refused to vote for me\n");
		return
	}

	n := len(rf.peers)
	numVotedForMe := len(rf.VotedForMe)

	quorum := (n / 2) + 1

	rf.Dbg("Election finished, numVotedForMe=%d, n=%d, quorum=%d\n",
		numVotedForMe, n, quorum)

	if numVotedForMe >= quorum {
		rf.Dbg("becoming a leader, rf.VotedForMe=%v\n", rf.VotedForMe)
		rf.State = LEADER
		rf.wakeupTicker()
	} else {
		rf.Dbg("pursuing to become a leader, numVotedForMe=%d, rf.VotedForMe=%v\n",
			numVotedForMe, rf.VotedForMe)
	}
}

func (rf *Raft) startNewElection() {
	//按照Raft论文描述，把自己的Term+1，状态设置为Candidate，然后发起
	//选举请求
	rf.CurrentTerm += 1
	rf.State = CANDIDATE

	rf.Dbg("start a new election\n")

	//先清空这个切片VotedForMe
	rf.VotedForMe = rf.VotedForMe[:0]

	//首先给自己投一票
	rf.VotedFor = rf.me
	rf.VotedForMe = append(rf.VotedForMe, rf.me)

	rf.persist()

	//循环遍历所有Raft Servers，给他们发送RequestVote消息
	n := len(rf.peers)
	for i := 0; i < n; i++ {
		//已经给自己投过票了，不用给自己发送RV消息了
		if i == rf.me {
			continue
		}

		//创建一个goroutine去执行发送RequestVote，并且等待reply。因为，
		//在网络故障的情况下，这个sendRequestVote函数可能耗时很长(2s)
		//这样。如果顺序执行，那么第1个节点故障，就会影响后续的发送。
		//用goroutine去发送RV并等待回复，多个goroutine可以一起并发执行。
		go rf.pleaseVoteForMe(i)
	}
}


//这个函数产生随机的ElectionTime时间
func (rf *Raft) genElectionTimeout() time.Duration {
	//产生一个150到350之间的随机数，这个随机数，就作为Election Timeout
	span := int64(ElectionTimeoutMax - ElectionTimeoutMin)
	ms := ElectionTimeoutMin + (rand.Int63() % span)

	duration := time.Duration(ms) * time.Millisecond
	return duration
}

//提前唤醒这个ticker线程的timer
func (rf *Raft) wakeupTicker() {
	rf.mu.ShouldBeLocked()
	rf.Dbg("wakeup, sleeping=%v\n", rf.sleeping)

	if rf.sleeping {
		rf.Timer.Reset(0)
	}
}

//这个函数在ticker线程(goroutine)中调用，睡眠一段时间。
//调用这个函数睡眠期间，可以调用wakeupTicker提前唤醒。
func (rf *Raft) sleep(duration time.Duration) {

	rf.Dbg("sleep in, duration=%v\n", duration)

	rf.mu.ShouldBeLocked()

	rf.assert(rf.sleeping == false)
	rf.sleeping = true
	rf.Timer = time.NewTimer(duration)

	rf.mu.Unlock()

	select {
	case <-rf.Timer.C:
	}
	rf.mu.Lock()

	rf.sleeping = false
	rf.Dbg("sleep out: duration=%v\n", duration)
}

func (rf *Raft) changeTerm(term int) {
	rf.CurrentTerm = term
	rf.NumKeepAlive++
	rf.persist()
}

func (rf *Raft) becomeFollower(term int) {
	rf.mu.ShouldBeLocked()
	rf.assert(rf.State != FOLLOWER)
	rf.CurrentTerm = term
	rf.State = FOLLOWER
	rf.NumKeepAlive = 1
	rf.wakeupTicker()
	rf.persist()
}

//在follower状态的时候，ticker调用这个函数
func (rf *Raft) followerTick() {

	//这里先获取mutex锁，获取到所之后，再访问rf.State，rf.NumKeepAlive等
	//变量。因为这些变量
	//rf.mu.Lock()
	//defer rf.mu.Unlock()

	rf.Dbg("before sleep, rf.NumKeepAlive=%d\n", rf.NumKeepAlive)
	et := rf.genElectionTimeout()

	rf.mu.Lock()
	rf.sleep(et)
	rf.mu.Unlock()

	rf.Dbg("after sleep, rf.NumKeepAlive=%d\n", rf.NumKeepAlive)

	//如果当前的状态，不是FOLLOWER状态，那么就退出
	if rf.State != FOLLOWER {
		rf.Dbg("I'm not a follower\n")
		return
	}

	//如果在睡眠Election Timeout时间内收到过AppendEntries消息，则说明
	//Leader是活的，返回，什么都不做
	if rf.NumKeepAlive > 0 {
		rf.Dbg("got keepalive\n")
		rf.NumKeepAlive = 0
		return
	}

	//如果当前状态是FOLLOWER，并且在Election Timeout时间内没有收到
	//AppenEntries消息，说明Leader已经死了。
	rf.startNewElection()
	rf.Dbg("exit from followerTick()\n")
}

//在candidate状态的时候，ticker调用这个函数
func (rf *Raft) candidateTick() {

	rf.Dbg("Candidate ticker\n")

	rf.mu.Lock()
	defer rf.mu.Unlock()

	et := rf.genElectionTimeout()
	rf.sleep(et)


	//如果当前的状态，不是candidate状态，那么就退出
	if rf.State != CANDIDATE {
		return
	}

	rf.startNewElection()
}

//把日志应用到状态机上
func (rf *Raft)Apply(index int, command interface{}) {
	s := SprintCmd(command)
	rf.Dbg("Apply, Index=%d, cmd=%s\n", index, s)
	am := &ApplyMsg {
		CommandValid: true,
		Command: command,
		CommandIndex: index,
	}
	rf.ApplyCh<- *am
	rf.LastApplied = index
}

//调用这个函数发送KeepAlive消息
func (rf *Raft) sendLogEntriesOrKeepAlive(to int) bool {

	//TODO: 检查如果没有新的日志需要发，并且keepalive定时器也没有到期
	//      则返回

	//构造AppendEntries消息，发给对方
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.State != LEADER {
		rf.Dbg("I am not leader longer\n");
		//rf.mu.Unlock()
		return false
	}

	args := &AppendEntriesArgs{}
	reply := &AppendEntriesReply{}
	args.Term = rf.CurrentTerm
	args.LeaderId = rf.me
	curr := rf.NextIndex[to]

	if curr > len(rf.Log) {
		curr = len(rf.Log)
	}

	prev := curr - 1

	rf.assert(prev >= 0)

	args.PrevLogIndex = prev
	args.PrevLogTerm = rf.Log[prev].Term
	args.LeaderCommit = rf.CommitIndex
	args.Entries = rf.Log[curr:]
	rf.aeTxSeq[to]++
	args.Seq = rf.aeTxSeq[to]
	rf.NextIndex[to] = len(rf.Log)
	//rf.mu.Unlock()

	sendInstallsnap:=false
	if(args.PrevLogIndex<rf.LastsnapIndex){
		sendInstallsnap = true
	}else if(rf.reachid(len(rf.Log)-1)>args.PrevLogTerm){
		args.Entries=rf.Log[rf.findid(args.PrevLogIndex+1):]
	}else {
		args.Entries=make([]LogEntry,0)
	}

	if(sendInstallsnap){
		go rf.handleInstallsnap(to)
		return false
	}

	//这里可能出现的问题是
	ok := rf.sendAppendEntries(to, args, reply)

	//rf.mu.Lock()
	//defer rf.mu.Unlock()

	if !ok {
		return false
	}

	//如果收到了Seq更新的AE reply，那么就丢弃就消息
	if reply.Seq < rf.aeRxSeq[to] {
		rf.Dbg("reply.seq=%d mismatch seq=%d\n",
			reply.Seq, rf.aeRxSeq[to])
		return false
	}

	if rf.State != LEADER {
		rf.Dbg("I am not leader longer\n");
		return false
	}

	if reply.Term > rf.CurrentTerm {
		rf.Dbg("My term is lagged\n");
		rf.VotedFor = -1
		rf.becomeFollower(reply.Term)
		return false
	}

	//如果返回Append失败，那么返回重新发送
	if !reply.Success { 
		i := reply.LastIndex + 1
		if i > len(rf.Log) {
			i = len(rf.Log)
		}
		rf.NextIndex[to] = i
		rf.MatchIndex[to] = 0
		rf.Dbg("reply from %d, NextIndex=%d\n", to, i)
		return true
	}

	rf.assert(reply.LastIndex <= len(rf.Log)-1)

	rf.MatchIndex[to] = reply.LastIndex
	//检查哪些可以commit了
	n := len(rf.peers)
	m := 0
	cur := rf.MatchIndex[to]
	for i:=0; i<len(rf.peers); i++ {
		if rf.MatchIndex[i] >= cur {
			m++
		}
	}
	quorum := (n / 2) + 1
	if (m >= quorum) && (rf.CommitIndex < cur) {
		if rf.Log[cur].Term == rf.CurrentTerm {
			rf.CommitIndex = cur
		}
	}
	for i:=rf.LastApplied+1; i<=rf.CommitIndex; i++ {
		rf.Apply(i, rf.Log[i].Command)
	}
	rf.aeRxSeq[to] = reply.Seq
	return false
}

func (rf *Raft) handleSingleFollower(i int) {

	for {
		retry := rf.sendLogEntriesOrKeepAlive(i)
		if !retry {
			break
		}
	}
}

//在leader状态的时候，ticker调用这个函数
func (rf *Raft) leaderTick() {
	duration := KeepAlivePeriod * time.Millisecond
	rf.Dbg("duration=%v\n", duration)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.sleep(duration)

	if rf.State != LEADER {
		rf.Dbg("lost leader\n")
		return
	}

	n := len(rf.peers)
	for i := 0; i < n; i++ {
		//不需要给自己发送keepalive消息
		if i == rf.me {
			continue
		}

		go rf.handleSingleFollower(i)
	}
}

//这是一个gorouting，用来执行一些定时任务
func (rf *Raft) ticker() {
	for rf.killed() == false {
		// Your code here (3A)
		// Check if a leader election should be started.
		//rf.Dbg("ticker: me=%d, state=%d, term=%d\n",
		//	rf.me, rf.State, rf.CurrentTerm)
		if rf.State == FOLLOWER {
			rf.followerTick()
		} else if rf.State == CANDIDATE {
			rf.candidateTick()
		} else if rf.State == LEADER {
			rf.leaderTick()
		}
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
	//DPrintf("Make %d\n", me)
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.State = FOLLOWER
	rf.ApplyCh = applyCh
	rf.VotedFor = -1
	rf.mu.Init(me)

	if me==0 {
		chLog = make(chan string, 1000)
		go dumpLog()
	}

	n := len(peers)
	rf.NextIndex = make([]int, n)
	rf.MatchIndex = make([]int, n)
	rf.aeTxSeq = make([]int, n)
	rf.aeRxSeq = make([]int, n)

	rf.Log = make([]LogEntry, 0)
	firstEntry := LogEntry{Term: -1, Index: -1, Command:-1 }
	rf.Log = append(rf.Log, firstEntry)

	for i:=0; i<n; i++ {
		rf.NextIndex[i] = 1
		rf.MatchIndex[i] = 0
	}

	// Your initialization code here (3A, 3B, 3C).

	// initialize from state persisted before a crash
	rf.ReadSnapshot(persister.ReadSnapshot())
	rf.readPersist(persister.ReadRaftState())

	for i:=0;i<len(rf.NextIndex);i++{
		rf.NextIndex[i]=rf.reachid(len(rf.Log))
	}


	// start ticker goroutine to start elections
	go rf.ticker()

	return rf
}



